import java.util.Properties
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types._

object jdbcrw {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("JDBC")
      .master("local")
      .config("spark.testing.memory", "2147480000")
      .getOrCreate()

    val jdbcDF = spark.read.format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/xiami")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("dbtable", "user_login")
      .option("user","root")
      .option("password","poiuyt")
      .load()

    jdbcDF.show()

    import spark.implicits

    val userRDD = spark.sparkContext.parallelize(Array("lpp@163.com,lpp,25376456,poiuytrewq",
                                                       "ff@126.cn,ff,2536476,lkjhgfdsa",
                                                       "xs@sina.com,xs,2435654,xspoiuyt")).map(_.split(","))
    val schema = StructType(List(StructField("email", StringType,true),
                            StructField("uname", StringType,true),
                            StructField("uid", StringType,true),
                            StructField("password", StringType,true)))
    val rowRDD = userRDD.map(p => (Row(p(0).trim, p(1).trim, p(2).trim, p(3).trim)))
    val userDF = spark.createDataFrame(rowRDD, schema)

    val prop = new Properties()
    prop.setProperty("user","root")
    prop.setProperty("password","poiuyt")
    prop.setProperty("driver", "com.mysql.jdbc.Driver")

    userDF.write.mode("append")
      .jdbc("jdbc:mysql://localhost:3306/xiami", "xiami.user_login", prop)

  }

}
